package com.atguigu.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.app.BaseApp2;
import com.atguigu.realtime.bean.OrderDetail;
import com.atguigu.realtime.bean.OrderInfo;
import com.atguigu.realtime.bean.OrderWide;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.util.DimUtil;
import com.atguigu.realtime.util.JDBCUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.time.Duration;
import java.util.Map;

import static com.atguigu.realtime.common.Constant.*;

/**
 * Created by RUI on 2021/6/28 19:53
 */
public class DwmOrderWideApp extends BaseApp2 {

    public static void main(String[] args) {
        new DwmOrderWideApp().init(3003, 1, "DwmOrderWideApp", "DwmOrderWideApp", TOPIC_DWD_ORDER_INFO, TOPIC_DWD_ORDER_DETAIL);
    }

    @Override
    public void run(StreamExecutionEnvironment env, Map<String, DataStreamSource<String>> streamMap) {
        // 1. 两个事实表进行join
        SingleOutputStreamOperator<OrderWide> orderWideStreamWithoutDims = factJoin(streamMap);

        // 2. join维度信息
        dimJoin(orderWideStreamWithoutDims);
    }

    private void dimJoin(SingleOutputStreamOperator<OrderWide> orderWideStreamWithoutDims) {
        orderWideStreamWithoutDims
            .map(new RichMapFunction<OrderWide, OrderWide>() {

                private Connection conn;

                @Override
                public void open(Configuration parameters) throws Exception {
                    conn = JDBCUtil.getJdbcConnection(PHOENIX_DRIVER, PHOENIX_URL);
                }

                @Override
                public OrderWide map(OrderWide orderWide) throws Exception {
                    // 1. 读取用户的维度信息
                    JSONObject userInfo = DimUtil.readDimFromPhoenix(conn, Constant.DIM_USER_INFO, orderWide.getUser_id());
                    orderWide.setUser_gender(userInfo.getString("GENDER"));
                    return orderWide;
                }
            })
            ;
    }

    private SingleOutputStreamOperator<OrderWide> factJoin(Map<String, DataStreamSource<String>> streamMap) {

        KeyedStream<OrderInfo, Long> orderInfoStream = streamMap
            .get(TOPIC_DWD_ORDER_INFO)
            .map(x -> JSON.parseObject(x, OrderInfo.class))
            .assignTimestampsAndWatermarks(
                WatermarkStrategy.
                    <OrderInfo>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                    .withTimestampAssigner((data, ts) -> data.getCreate_ts())
            )
            .keyBy(OrderInfo::getId);

        KeyedStream<OrderDetail, Long> orderDetailStream = streamMap
            .get(TOPIC_DWD_ORDER_DETAIL)
            .map(x -> JSON.parseObject(x, OrderDetail.class))
            .assignTimestampsAndWatermarks(
                WatermarkStrategy.
                    <OrderDetail>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                    .withTimestampAssigner((data, ts) -> data.getCreate_ts())
            )
            .keyBy(OrderDetail::getOrder_id);

        return orderInfoStream
            .intervalJoin(orderDetailStream) //必须是 keyBy之后的流才可以interval join
            .between(Time.minutes(-5), Time.minutes(5)) // 指定上下界
            .process(new ProcessJoinFunction<OrderInfo, OrderDetail, OrderWide>() {
                @Override
                public void processElement(OrderInfo left, OrderDetail right, Context ctx, Collector<OrderWide> out) throws Exception {
                    out.collect(new OrderWide(left, right));
                }
            });
    }
}
